package com.s3u4.demo.zookeeper.service;

import com.s3u4.demo.zookeeper.ZkConfig;
import com.s3u4.demo.zookeeper.base.ConnectionWatcher;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * Created by Captain on 02/08/2017.
 */
public class ActiveKeyValueStore extends ConnectionWatcher {

    private static final String focus_key = "/demo-zookeeper/node1";

    public void read(){
        try {
            byte[] value = zooKeeper.getData(focus_key,this,null);
            System.out.println("获取到的值为 : " + new String(value,ZkConfig.CHARSET));
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        System.out.println("process with watcher status :" + watchedEvent.getState() );
        if ( watchedEvent.getState() == Event.KeeperState.SyncConnected ){
            connectSinal.countDown();
        }


        if ( watchedEvent.getType() == Event.EventType.NodeDataChanged ){
            read();
        }

    }

    /**
     * 设置 znode 值
     * @param path
     * @param value
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void write(String path,String value) throws KeeperException, InterruptedException {
        Stat stat = zooKeeper.exists(path,false);
        if ( stat == null ){
            zooKeeper.create(path,value.getBytes(ZkConfig.CHARSET), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } else {
            zooKeeper.setData(path,value.getBytes(ZkConfig.CHARSET), -1);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

        new Thread( () -> {
            Random random = new Random();
            try {
                ActiveKeyValueStore watcher = new ActiveKeyValueStore();
                watcher.connect(ZkConfig.ZK_HOST);

                watcher.read();

                while (true){
                    String value = random.nextInt(100) + "";

                    watcher.write(focus_key, value);
                    System.out.println("设置值为 : " + value);
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();


    }


}
